package com.intct.flink.project.dim;

import com.intct.common.FlinkSqlWithUtil;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-10-16 09:05
 */
public class DimOrderStateSQLApp2 {
    public static void main(String[] args) throws Exception {
        // 创建Stream环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建Table环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 设置全局并行度
        env.setParallelism(1);

        // 开启检查
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);

        // 映射MySQL订单状态表
        tenv.executeSql(
                "CREATE TABLE order_state(                " +
                        "     json_data        STRING              " +
                        " ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc'            " +
                        "     ,'hostname' = 'cdh-node'             " +
                        "     ,'port' = '13306'                    " +
                        "     ,'username' = 'root'                 " +
                        "     ,'password' = 'Test_090110'          " +
                        "     ,'database-name' = 'travel'          " +
                        "     ,'table-name' = '.*'                 " +
                        "     ,'scan.incremental.snapshot.enabled' = 'true'          " +
                        "     ,'debezium.schema.changes.include' = 'true'          " +
                        "     ,'format' = 'debezium-json'          " +
                        " )                                        "
        );

        tenv.executeSql("select * from order_state").print();

    }
}
